package com.meihua.code.thread;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.qingqing.common.util.JsonUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @program: full-crmsvc
 * @description:
 * @author: meihua
 * @created: 2021/09/22 14:10
 */
public class ElasticThreadPoolExecutor {

    private final ThreadPoolExecutor poolExecutor;

    private static final Logger logger = LoggerFactory.getLogger(ElasticThreadPoolExecutor.class);

    private  int capacity;
    private  int corePoolSize;
    private  int maximumPoolSize;
    private  boolean enableSyncSending =true;
    private  boolean isEnableElastic =true;
    private String poolName;


    /**
     * 弹性分数
     **/
    private Map<Integer,Integer> elasticScoreMap;

    public ElasticThreadPoolExecutor(
                                     String  poolName,
                                     int corePoolSize,
                                     int maximumPoolSize,
                                     long keepAliveTime,
                                     TimeUnit unit,
                                     BlockingQueue<Runnable> workQueue) {
        this.poolName=poolName;
        this.corePoolSize=corePoolSize;
        this.maximumPoolSize=maximumPoolSize;
        this.capacity=workQueue.size();
        this.poolExecutor =  new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                new ThreadFactoryBuilder().setNameFormat(poolName+"-pool-%d").build(),
                (runnable,executor)->{
                    try {
                        elastic();
                        if (enableSyncSending){
                            runnable.run();
                            logger.warn("mq msg queue is full. sender mq message sync :{}", JsonUtil.getJsonFromObject(runnable));
                        }
                    }catch (Exception e){
                        logger.error("mq rejected execution handler error ",e);
                    }
                });

        start();

    }

    private void start(){
        if (this.corePoolSize==this.maximumPoolSize ){
            this.isEnableElastic =false;
        }
        if (!isEnableElastic){
            return;
        }
        poolExecutor.allowCoreThreadTimeOut(true);
        elasticScoreMap = new HashMap<>(maximumPoolSize);
        for (int i=0;i<=maximumPoolSize;i++){
            elasticScoreMap.put(i,(capacity/maximumPoolSize)*i);
        }
        logger.info("elasticScoreMap: {}",elasticScoreMap);
    }




    public void execute(Runnable runnable){
        elastic();
        if (poolExecutor.isShutdown()){
            throw new RuntimeException("poolExecutor isShutdown !");
        }
        poolExecutor.execute(runnable);
    }

    private void elastic(){
        if (!isEnableElastic){
            return;
        }
        //当前弹性分数
        Integer elasticScore = elasticScoreMap.get(corePoolSize);
        int size = poolExecutor.getQueue().size();
        if (elasticScore<size&&corePoolSize<maximumPoolSize){
            logger.info("elasticScore:{} ,size: {}  add Thread",elasticScore,size ,poolExecutor.getPoolSize());
            corePoolSize++;
            poolExecutor.setCorePoolSize(corePoolSize);
        }else if (elasticScore>size&&corePoolSize>0){
            corePoolSize--;
            poolExecutor.setCorePoolSize(corePoolSize);
            logger.info("降级");
        }
    }

   public void getInfo(){
       logger.info("{}-pool-monitor:  PoolSize: {}, CorePoolSize: {}, Active: {}, Completed: {}, Task: {}, Queue: {}, LargestPoolSize: {}, MaximumPoolSize: {},  KeepAliveTime: {}, isShutdown: {}, isTerminated: {}",
               this.poolName,
               this.poolExecutor.getPoolSize(), this.poolExecutor.getCorePoolSize(), this.poolExecutor.getActiveCount(),
               this.poolExecutor.getCompletedTaskCount(), this.poolExecutor.getTaskCount(), this.poolExecutor.getQueue().size(), this.poolExecutor.getLargestPoolSize(),
               this.poolExecutor.getMaximumPoolSize(), this.poolExecutor.getKeepAliveTime(TimeUnit.MILLISECONDS), this.poolExecutor.isShutdown(), this.poolExecutor.isTerminated());
   }

   public void  shutdown(){
        poolExecutor.shutdown();
    }

}
